package com.huwaiwai.disruptor.demo01;

import com.huwaiwai.thread.thread.NameThreadFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.junit.Before;
import org.junit.Test;

import java.nio.ByteBuffer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 测试类
 * Created by huweiwei on 2017/4/11.
 */
@SuppressWarnings(value = {"unchecked"})
public class LongEventTest {

    /**
     * 默认流大小
     */
    private static final int DEFAULT_BUFFER_SIZE = 1024;
    /**
     * 循环发布次数
     */
    private static final long FOR_TIME = 5;

    private int bufferSize;
    private LongEventFactory factory;
    private Disruptor<LongEvent> disruptor;

    /**
     * 初始化
     */
    @Before
    public void init(){
        bufferSize = DEFAULT_BUFFER_SIZE;
        factory = new LongEventFactory();
        disruptor = new Disruptor<>(factory, bufferSize, new NameThreadFactory("Disruptor-Thread", true));
    }

    public void publish(){
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        LongEventProducer producer = new LongEventProducer(ringBuffer);
        ByteBuffer byteBuffer = ByteBuffer.allocate(8);
        for (long l = 0; l < FOR_TIME; l++) {
            byteBuffer.putLong(0, l);
            producer.onData(byteBuffer);
        }
    }

    /**
     * 配置两个处理器
     * 同一个流按序执行
     * @throws InterruptedException
     */
    @Test
    public void test01() throws InterruptedException {
        final CountDownLatch latch = new CountDownLatch(10);
        EventHandler<LongEvent> event01 = (LongEvent longEvent, long l, boolean b) -> {
            System.out.println(Thread.currentThread().getName() + "执行event01-" + longEvent.getValue());
            latch.countDown();
        };
        EventHandler<LongEvent> event02 = (LongEvent longEvent, long l, boolean b) -> {
            System.out.println(Thread.currentThread().getName() + "执行event02-" + longEvent.getValue());
            latch.countDown();
        };

        disruptor.handleEventsWith(event01)
                .then(event02);
        disruptor.start();
        publish();
        if (!latch.await(20, TimeUnit.SECONDS)) {
            System.out.println("latch GG");
        }
    }

    @Test
    public void test02() throws InterruptedException {
    }
}
